package org.bing.zion.core;

import java.io.IOException;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;

import org.bing.engine.common.logging.Log;
import org.bing.engine.common.logging.LogFactory;
import org.bing.zion.helper.ChannelBuilder;
import org.bing.zion.helper.SelectorBuilder;


public class Acceptor implements Runnable {
	private static final Log logger = LogFactory.getLog(Acceptor.class);
	private static final int DEFAULT_SELECT_TIMEOUT = 1000;
	private static final int coreNum = Runtime.getRuntime().availableProcessors();
	protected Selector selector;
	protected ServerSocketChannel ssChannel;
	protected ProtocolFilterChain filterChain;
	protected MessageHandlerChain handlerChain;
	protected ProcessorPool processorPool;

	/** processor num, performer num */
	public Acceptor(String ip, int port) throws IOException {
		this(ip, port, coreNum, coreNum * 2);
	}

	public Acceptor(String ip, int port, int processorNum, int performerNum) throws IOException {
		// init filter and handler, dispatcher
		filterChain = new ProtocolFilterChain();
		filterChain.setMessageDispatcher(new MulThreadDispatcher(performerNum));
		handlerChain = new MessageHandlerChain();
		// init channel and selector, processor
		ssChannel = ChannelBuilder.newServerSocketChannel(ip, port);
		processorPool = new ProcessorPool(processorNum);
		selector = SelectorBuilder.newSelector();
		ssChannel.register(selector, SelectionKey.OP_ACCEPT);
	}

	public void addProtocolFilter(ProtocolFilter filter) {
		filterChain.addFilter(filter);
	}

	public void addMessageHandler(MessageHandler handler) {
		handlerChain.addHandler(handler);
	}

	public void accept(SelectionKey key) throws IOException {
		if (key.isValid() && key.isAcceptable()) {
			ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
			SocketChannel sc = ssc.accept();
			sc.configureBlocking(false);
			logger.info("Accept " + sc.socket());
			Processor processor = processorPool.nextProcessor();
			Session session = new Session(processor, sc, filterChain, handlerChain);
			processor.read(session);
		} else {
			logger.error("Key " + key + " state error!");
		}
	}

	public void run() {
		Thread.currentThread().setName("seda-acceptor-main");
		for (;;) {
			try {
				int state = selector.select(DEFAULT_SELECT_TIMEOUT);
				if (state > 0) {
					for (Iterator<SelectionKey> iterator = selector.selectedKeys().iterator(); iterator.hasNext();) {
						SelectionKey key = (SelectionKey) iterator.next();
						iterator.remove();
						try {
							accept(key);
						} catch (IOException e) {
							logger.error("Fail to handle accept!", e);
							close(key.channel());
						}
					}
				}
			} catch (Exception e) {
				logger.error("Failed in main loop!", e);
			}
		}
	}

	private void close(SelectableChannel channel) {
		try {
			if (channel != null && channel.isOpen()) {
				channel.close();
			}
		} catch (IOException e) {
			logger.error("Fail to close channel " + channel, e);
		}
	}
}
